<?php
declare (strict_types=1);

namespace job;

use app\admin\model\base\SysLog;
use app\admin\model\iot\DeviceApkLog;
use iot\DeviceClient;
use think\facade\Log;
use think\facade\Queue;
use think\queue\Job;

/**
 * ThinkQueueJob队列
 */
class ThinkQueueJob
{
    //当轮到该任务时，系统将生成一个该类的实例,并调用其receive方法
    public static string $jobHandlerClassName = 'job\ThinkQueueJob@receive';
    //当前任务归属的队列名称,如果为新队列,会自动创建
    public static string $jobQueueName = "ThinkQueueJob";

    /**
     * 发送队列至thinkQueue
     * @param string $type
     * @param string $topic
     * @param mixed $msg
     * @param int $later
     * @return false|mixed|void|null
     */
    public static function sendMsg(string $type, string $topic, mixed $msg, int $later = 0)
    {
        $jobData = [
            'type' => $type,
            'topic' => $topic,
            'msg' => $msg,
            'time' => time(),
        ];
        //将任务推送到消息队列,等待对应的消费者去执行
        if ($later) {
            $isPushed = Queue::later($later, self::$jobHandlerClassName, $jobData, self::$jobQueueName);
        } else {
            $isPushed = Queue::push(self::$jobHandlerClassName, $jobData, self::$jobQueueName);
        }
        //redis驱动时返回值为随机字符串|false
        if ($isPushed === false) {
            Log::log('alert', "Oops, something went wrong.");
        }
        return $isPushed;
    }

    /**
     * 任务失败
     * @param $data
     * @return void
     */
    public function failed($data)
    {
        var_dump($data);
    }

    /**
     * ThinkQueue消费者
     * @param Job $job
     * @param mixed $data
     * @return void
     */
    public function receive(Job $job, mixed $data)
    {
        //如果任务执行成功,记得删除任务
        $job->delete();
        //根据消息中的数据进行实际的业务处理
        if (is_array($data)) {
            self::consumerProgram($data);
        }
        if ($job->attempts() > 3) {
            //通过这个方法可以检查这个任务已经重试了几次了
            Log::log('alert', "dismiss job has been retried more that 3 times");
            $job->delete();
            //也可以重新发布这个任务
            //$delay为延迟时间，表示该任务延迟2秒后再执行
            $job->release(2);
        }
    }

    /**
     * 执行消费程序分发窗口
     * @param array $data
     * @return void
     */
    private static function consumerProgram(array $data)
    {
        $run = DeviceClient::getInstance();
        //根据类型走不同流程
        switch ($data['type']) {
            //写入系统日志
            case 'WriteSystemLog':
                SysLog::writeSysLog($data['msg']);
                break;
            //更新系统日志
            case 'UpdateSystemLog':
                SysLog::updateSysLog($data['msg']);
                break;
            //处理订单队列
            case 'ProcessOrderQueue':
                $run->processOrderQueue($data['msg']);
                break;
            //MQTT客户端设备上报
            case 'EmqxMqttClient':
                $run->clientMessage($data['msg'], $data['time']);
                break;
            //MQTT节点上线系统消息
            case 'EmqxSysMsgConnected':
                $run->sysMsgConnected($data);
                break;
            //MQTT节点离线系统消息
            case 'EmqxSysMsgDisconnected':
                $run->sysMsgDisconnected($data);
                break;
            //V2系统设备APK升级
            case 'V2DeviceApkRelease':
                DeviceApkLog::writeApkUpgrade($data['msg']);
                break;
            //V2取袋同步订单至V1
            case 'SynOrderV2ToV1':
                $run->synOrderV2ToV1($data['msg']);
                break;
            default:
                break;
        }
    }

}
